package dshumsky.akka;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import akka.actor.UntypedActor;


public class AggregateActor
    extends UntypedActor {
    private Map<String, Integer> finalReducedMap = new HashMap<String, Integer>();

    @Override
    public void onReceive(Object message)
        throws Exception {

        if (message instanceof ReduceData) {
            ReduceData reduceData = (ReduceData) message;
            aggregateInMemoryReduce(reduceData.getReduceDataList());
        } else if (message instanceof Result) {
            getSender().tell(finalReducedMap.toString(), getSelf());
        } else {
            unhandled(message);
        }
    }

    private void aggregateInMemoryReduce(Map<String, Integer> reducedList) {

        for (Entry<String, Integer> e : reducedList.entrySet()) {
            Integer sum = finalReducedMap.get(e.getKey());

            if (sum == null) {
                finalReducedMap.put(e.getKey(), e.getValue());
            } else {
                finalReducedMap.put(e.getKey(), sum + e.getValue());
            }
        }
    }

}
